package com.atguigu.gmall.realtime.common.util;

import com.alibaba.fastjson.JSONObject;
import com.google.common.base.CaseFormat;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Set;

/**
 * ClassName: HBaseUtil
 * Description:
 *
 * @Create 2024/4/16 15:44
 */
public class HBaseUtil {

    //TODO 获取连接
    //同步
    public static Connection getHBaseConnection() {
        try {
            Configuration conf = new Configuration();
            conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
            Connection connection = ConnectionFactory.createConnection(conf);
            return connection;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }

    //异步
    public static AsyncConnection getAsyncHBaseConnection() {
        try {
            Configuration conf = new Configuration();
            conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
            AsyncConnection asyncConnection = ConnectionFactory.createAsyncConnection(conf).get();
            return asyncConnection;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    //TODO 关闭连接
    //同步
    public static void closeHBaseConn(Connection hbaseConn) throws IOException {
        if (hbaseConn != null && !hbaseConn.isClosed()) {
            hbaseConn.close();
        }
    }

    //异步
    public static void closeAsyncHBaseConn(AsyncConnection asyncConnection) throws IOException {
        if (asyncConnection != null && !asyncConnection.isClosed()) {
            asyncConnection.close();
        }
    }


    //TODO 建表
    public static void createHBaseTable(Connection hbaseConn, String nameSpace, String table, String... families) throws IOException {

        if (families.length < 1) {
            System.out.println("必须指定列族");
            return;
        }

        //try(xxx)自动资源释放
        try (Admin admin = hbaseConn.getAdmin()) {

            TableName tableName = TableName.valueOf(nameSpace, table);

            // 判断要建的表是否存在
            if (admin.tableExists(tableName)) {
                System.out.println("要创建的" + nameSpace + "下的表" + table + "已经存在");
                return;
            }

            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);

            for (String family : families) {
                ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
            }

            admin.createTable(tableDescriptorBuilder.build());
            System.out.println("创建" + nameSpace + "下的表" + table);

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    //TODO 删表
    public static void dropHBaseTable(Connection hbaseConn, String nameSpace, String table) throws IOException {

        try (Admin admin = hbaseConn.getAdmin()) {

            TableName tableName = TableName.valueOf(nameSpace, table);

            // 判断要建的表是否存在
            if (!admin.tableExists(tableName)) {
                System.out.println("要删除的" + nameSpace + "下的表" + table + "不存在");
                return;
            }

            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            System.out.println("删除" + nameSpace + "下的表" + table);

        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }

    /**
     * TODO put数据
     *
     * @param hbaseConn      连接对象
     * @param namespace      表空间
     * @param tableName      表名
     * @param rowKey         主键
     * @param family         列族
     * @param dataJsonObject 要写入的json对象
     */
    public static void putRow(Connection hbaseConn, String namespace, String tableName, String rowKey, String family, JSONObject dataJsonObject) {
        TableName tableNameObj = TableName.valueOf(namespace, tableName);

        try (Table table = hbaseConn.getTable(tableNameObj)) {

            Put put = new Put(Bytes.toBytes(rowKey));

            Set<String> columns = dataJsonObject.keySet();

            for (String column : columns) {
                String value = dataJsonObject.getString(column);
                if (value != null) {
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), Bytes.toBytes(value));
                }
            }

            table.put(put);

            System.out.println("向" + namespace + "下的表" + tableName + "插入" + rowKey + "数据成功");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    //TODO 删除数据
    public static void delRow(Connection hbaseConn, String namespace, String tableName, String rowKey) {
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        try (Table table = hbaseConn.getTable(tableNameObj)) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
            System.out.println("删除" + namespace + "下的表" + tableName + "中的" + rowKey + "数据成功");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //TODO 根据rowKey从表中获取数据
    //同步
    public static <T> T getRow(Connection hbaseConn, String namespace, String tableName, String rowKey, Class<T> tClass, boolean... isUnderlineToCamel) {
        boolean defaultIsUToC = false;  // 默认不执行下划线转驼峰

        if (isUnderlineToCamel.length > 0) {
            defaultIsUToC = isUnderlineToCamel[0];
        }
        TableName tableNameObj = TableName.valueOf(namespace, tableName);

        try (Table table = hbaseConn.getTable(tableNameObj)) {

            Get get = new Get(Bytes.toBytes(rowKey));

            Result result = table.get(get);

            List<Cell> cellList = result.listCells();

            if (cellList != null && cellList.size() > 0) {

                T obj = tClass.newInstance();

                for (Cell cell : cellList) {
                    //cell的主键由{rowkey, column Family：column Qualifier, timestamp}组成，方法返回的都是同一个数组对象
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));

                    if (defaultIsUToC) { // 需要下划线转驼峰:  a_a => aA a_aaaa_aa => aAaaaAa
                        columnName = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, columnName);
                    }

                    BeanUtils.setProperty(obj, columnName, columnValue);

                }

                return obj;
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        return null;

    }

    //异步
    public static JSONObject getRowAsync(AsyncConnection asyncConnection, String namespace, String tableName, String rowKey) {
        TableName tableNameObj = TableName.valueOf(namespace, tableName);

        //asyncTable对象不需要关闭，由于是异步处理，不关闭可以去做别的处理，等回调函数返回结果，底层会自动关闭
        AsyncTable<AdvancedScanResultConsumer> asyncTable = asyncConnection.getTable(tableNameObj);

        Get get = new Get(Bytes.toBytes(rowKey));

        try {
            Result result = asyncTable.get(get).get();

            List<Cell> cellList = result.listCells();

            if (cellList != null && cellList.size() > 0) {
                JSONObject jsonObj = new JSONObject();
                
                for (Cell cell : cellList) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName, columnValue);
                }

                return jsonObj;

            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        
        return null;
        
    }

}
